package com.longtengyuyun.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author ZhaoYunLong
 * @date 2022/3/11
 * @description
 */
public class KafkaCustomConsumer {

    public static void main(String[] args) {
        //        创建Consumer
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.10.60.119:9092,10.10.60.120:9092,10.10.60.121:9092");
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        kafkaConsumer.subscribe(Arrays.asList("topic_zyl"));

        while (true) {

            kafkaConsumer.beginningOffsets();
        }

        for (int i = 0; i < 10; i++) {
            ProducerRecord topic_zyl = new ProducerRecord("topic_zyl", "zyl" + i, "zyl_value" + i);
            kafkaProduct.send(topic_zyl);
        }
    }


}
